package commands;

import java.io.IOException;

import mapReduce.TopTenPhotosMR;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Job;

import vo.Vote;



public class TopTenPhotosCommand implements Comando {

	public String execute(String [] args) throws IOException{
		
		if (args.length != 1) {
			return getSintaxe();
		}

		Configuration conf = HBaseConfiguration.create();
		
		try {
			createJob(conf).waitForCompletion(true);
		} catch (InterruptedException e) {
			e.printStackTrace();
			
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
			
		}		
			
		return TopTenPhotosMR.Reducer.printTopTenPhotos();
	}

	public String getSintaxe() {
		return "-topTenPhotos";
	}
	
	public Job createJob(Configuration conf) throws IOException {
		Job job = new Job(conf, "teste");
		job.setJarByClass(this.getClass());

		String table = Vote.TABLENAME;

		Scan scan = new Scan();
		scan.addColumn(Bytes.toBytes(Vote.CF_ATTRIBUTES),
				Bytes.toBytes(Vote.VOTE_VALUE));

		job.setNumReduceTasks(1);

		TableMapReduceUtil
				.initTableMapperJob(table, scan, TopTenPhotosMR.Mapper.class,
						ImmutableBytesWritable.class,
						ImmutableBytesWritable.class, job);

		TableMapReduceUtil
				.initTableReducerJob(table, TopTenPhotosMR.Reducer.class, job);

		return job;
	}

	@Override
	public String getKey() {
		return "-topTenPhotos";
	}

}